/*******************************************************************************
 * * Copyright 2012 Impetus Infotech.
 *  *
 *  * Licensed under the Apache License, Version 2.0 (the "License");
 *  * you may not use this file except in compliance with the License.
 *  * You may obtain a copy of the License at
 *  *
 *  *      http://www.apache.org/licenses/LICENSE-2.0
 *  *
 *  * Unless required by applicable law or agreed to in writing, software
 *  * distributed under the License is distributed on an "AS IS" BASIS,
 *  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  * See the License for the specific language governing permissions and
 *  * limitations under the License.
 ******************************************************************************/
package cn.ilikes.tools.hbase.jpa.schemas;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.slf4j.Logger;

import cn.ilikes.tools.hbase.jpa.JPAHbasePersistenUnitEnv;
import cn.ilikes.tools.hbase.jpa.SysConfigs;
import cn.ilikes.tools.hbase.jpa.clients.admins.JPAHbaseAdmin;
import cn.ilikes.tools.hbase.jpa.clients.datasource.HbaseDataSource;
import cn.ilikes.tools.hbase.jpa.metadatas.IEntityMetaData;
import cn.ilikes.tools.hbase.jpa.metadatas.IEntityMetaDataFactory;
import cn.ilikes.tools.hbase.jpa.metadatas.elements.EntityFields;
import cn.ilikes.tools.hbase.jpa.metadatas.sub.configs.Datastorefactory.DataStore;
import cn.ilikes.tools.hbase.jpa.schemas.processors.ISchemasProcessor;
import cn.ilikes.tools.hbase.jpa.schemas.processors.SchemasProcessor;
import cn.ilikes.tools.hbase.jpa.schemas.processors.TableFieldSchemasProcessor;
import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.untils.HBaseUtils;

public class HBaseSchemaManager implements SchemaManager {

	private Logger logger = Loggerfactory.instance(HBaseSchemaManager.class);

	private JPAHbaseAdmin jpaHbaseAdmin;

	private JPAHbasePersistenUnitEnv hbasePersistenUnitEnv;

	public HBaseSchemaManager(JPAHbasePersistenUnitEnv hbasePersistenUnitEnv) {
		this.hbasePersistenUnitEnv = hbasePersistenUnitEnv;

	}

	private JPAHbaseAdmin getJpaHbaseAdmin() {
		return jpaHbaseAdmin;
	}

	private void setJpaHbaseAdmin(JPAHbaseAdmin jpaHbaseAdmin) {
		this.jpaHbaseAdmin = jpaHbaseAdmin;
	}

	private ExecType ceateHbaseAdmin(ExecType execType) {
		if (execType.equals(ExecType.no))
			return execType;
		DataStore dataStore = hbasePersistenUnitEnv.gethBasePersistenceUnitMetadata().getDataStore();
		HbaseDataSource hbaseDataSource = new HbaseDataSource(dataStore);
		JPAHbaseAdmin hbaseAdmin = new JPAHbaseAdmin(hbaseDataSource);
		this.setJpaHbaseAdmin(hbaseAdmin);
		return execType;
	}

	private ISchemasProcessor initcheckSchemaTable() {
		List<ISchemasProcessor> schemasProcessors = new ArrayList<ISchemasProcessor>();
		schemasProcessors.add(new TableFieldSchemasProcessor());

		ISchemasProcessor schemasProcessor = new SchemasProcessor(schemasProcessors);
		return schemasProcessor;

	}

	public void checkSchemaTable() {
		ISchemasProcessor schemasProcessor = initcheckSchemaTable();
		schemasProcessor.processorTableSchemas(hbasePersistenUnitEnv);
	}

	@Override
	public void execSchema(ExecType execType) throws JPAHBaseSchemaException {
		if (!hbasePersistenUnitEnv.isInitComplete()) {
			throw new JPAHBaseSchemaException("this pu name : " + hbasePersistenUnitEnv.getPersistenceUnitsName() + " not init ! ");
		}
		checkSchemaTable();
		execType = ceateHbaseAdmin(execType);
		switch (execType) {
		case create: {
			create();
			break;
		}
		}

	}

	private void create() {
		HBaseAdmin hBaseAdmin = getJpaHbaseAdmin().getAdmin();

		IEntityMetaDataFactory entityMetaDataFactory = hbasePersistenUnitEnv.getEntityMetaDataFactory();

		Set<IEntityMetaData> entityMetaDatas = entityMetaDataFactory.listAllIEntityMetaData();
		//		System.out.println(" ============================================  " + entityMetaDatas.size());
		for (IEntityMetaData entityMetaData : entityMetaDatas) {
			HTableDescriptor tableDescriptor = getTableMetaData(entityMetaData);
			try {
				if (hBaseAdmin.tableExists(tableDescriptor.getName())) {
					logger.info(" table " + entityMetaData.getEntityTableName() + "  exists ! ");
					try {
						hBaseAdmin.getTableDescriptor(tableDescriptor.getName());
						hBaseAdmin.disableTable(tableDescriptor.getName());
						hBaseAdmin.deleteTable(tableDescriptor.getName());
					} catch (TableNotFoundException e) {
						logger.info(" table " + entityMetaData.getEntityTableName() + " not  exists ! ", e.getMessage());
						throw new JPAHBaseSchemaRuntimeException(e);
					} catch (IOException ioex) {
						logger.error("Either table isn't in enabled state or some network problem, Caused by: " + ioex.getMessage());
						//						throw new JPAHBaseSchemaRuntimeException(ioex);
						hBaseAdmin.deleteTable(tableDescriptor.getName());
					}
				}

				HTableDescriptor hTableDescriptor = getTableMetaData(entityMetaData);
				try {
					hBaseAdmin.createTable(hTableDescriptor);
				} catch (IOException ioex1) {
					Loggerfactory.error(logger, "Table isn't in enabled state, Caused by:" + ioex1.getMessage(), ioex1);
				}
			} catch (IOException ioex1) {
				Loggerfactory.error(logger, "Table isn't in enabled state, Caused by:" + ioex1.getMessage(), ioex1);
			}

		}
	}

	private HTableDescriptor getTableMetaData(IEntityMetaData entityMetaData) {
		Loggerfactory.info(logger, "init   table  Info  descriptor   table name : " + entityMetaData.getEntityTableName());
		HTableDescriptor hTableDescriptor = new HTableDescriptor(entityMetaData.getEntityTableName());
		Set<Map.Entry<String, EntityFields>> entrySets = entityMetaData.getEntityFields().entrySet();

		for (Entry<String, EntityFields> entrySet : entrySets) {
			EntityFields entityFields = entrySet.getValue();
			{

				if (entityFields.getjPAField() == null)
					continue;
				String columnName;

				if (entityFields.getjPAField().isRelation() && entityFields.getjPAField().getRelation().isColumnName()) {
					columnName = entityFields.getjPAField().getRelation().getColumnName();
					//          				int index = columnName.indexOf("@");
					//          				String family = HBaseUtils.getfamily(columnName, index);
					//          				Loggerfactory.info(logger, "init table column => columnName : " + columnName + "  familyName : " + family);
					//          				HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
					//          				hTableDescriptor.addFamily(hColumnDescriptor);
				} else {
					columnName = entityFields.getjPAField().getColumnName();
					//          				int index = columnName.indexOf("@");
					//          				String family = HBaseUtils.getfamily(columnName, index);
					//          				Loggerfactory.info(logger, "init table column => columnName : " + columnName + "  familyName : " + family);
					//          				HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
					//          				hTableDescriptor.addFamily(hColumnDescriptor);
				}
				if (columnName == null || columnName.isEmpty())
					continue;
				int index = columnName.indexOf("@");
				String family = HBaseUtils.getfamily(columnName, index);
				Loggerfactory.info(logger, "init table column => columnName : " + columnName + "  familyName : " + family);
				HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
				hTableDescriptor.addFamily(hColumnDescriptor);

			}
		}
		return hTableDescriptor;
	}

	@Override
	public ExecType getExecType() {
		String ddl_auto_prepare = (String) hbasePersistenUnitEnv.gethBasePersistenceUnitMetadata().getProperties().get(SysConfigs.jpa_datastore_ddl_auto_prepare);
		return ExecType.valueOf(ddl_auto_prepare == null ? "no" : ddl_auto_prepare.toLowerCase());
	}

}
